package compressor

import (
	"bytes"
	"fmt"
	"math"
	"strings"

	pb "gitee.com/shipnet/flogo_compressor/pb"
	ds "gitee.com/shipnet/flogo_edge_channel_data_store"

	"github.com/project-flogo/core/activity"
	"github.com/tidwall/gjson"
	"github.com/ulikunitz/xz"
)

type Activity struct {
	settings settings
}

func init() {
	_ = activity.Register(&Activity{}, New) //activity.Register(&Activity{}, New) to create instances using factory method 'New'
}

type settings struct{}

func New(ctx activity.InitContext) (activity.Activity, error) {
	st := settings{}
	act := &Activity{
		settings: st,
	}
	return act, nil
}

var activityMd = activity.ToMetadata(&Settings{}, &Input{}, &Output{})

func (a *Activity) Metadata() *activity.Metadata {
	return activityMd
}

func (a *Activity) Eval(ctx activity.Context) (bool, error) {
	input := &Input{}
	err := ctx.GetInputObject(input)
	if err != nil {
		return true, err
	}
	var buf []byte
	result := CompressResult{
		Encoder:          input.Encoder,
		EncoderParams:    input.EncoderParams,
		Compressor:       input.Compressor,
		CompressorParams: input.CompressorParams,
	}
	switch input.Encoder {
	case "JSON":
		{
			buf, err = a.evalJSON(ctx, input)
		}
	case "PB":
		{
			buf, err = a.evalPB(ctx, input)
		}
	default:
		{
			err = fmt.Errorf("unknown action: %s", input.Encoder)
		}
	}

	if err != nil {
		return true, err
	}

	switch input.Compressor {
	case "XZ":
		{
			var compressed []byte
			compressed, err = compress(buf)
			if err != nil {
				result.Data = buf
				result.Compressor = "NONE"
			} else {
				result.Data = compressed
			}
			break
		}
	default:
		{
			result.Data = buf
			err = nil
			break
		}
	}

	errStr := ""
	if err != nil {
		errStr = err.Error()
	}
	result.Err = errStr
	output := &Output{
		Result: result,
	}

	err = ctx.SetOutputObject(output)
	return true, err
}

func (a *Activity) evalJSON(ctx activity.Context, input *Input) ([]byte, error) {
	lst := make([]string, len(input.Data))
	for idx, d := range input.Data {
		p := d.Serialize()
		lst[idx] = p
	}
	b := "[" + strings.Join(lst, ",") + "]"
	return []byte(b), nil
}

func (a *Activity) evalPB(ctx activity.Context, input *Input) ([]byte, error) {
	edges := make([]*pb.EdgeChannelData, len(input.Data))
	for index, d := range input.Data {
		pd := &pb.EdgeChannelData{
			EdgeChannelId: d.ID,
			Timestamp:     d.Timestamp,
		}
		setValue(pd, d)
		edges[index] = pd
	}
	c := pb.Compression{
		Data: edges,
	}
	b := []byte{}
	b, err := c.XXX_Marshal(b, false)
	if err != nil {
		return []byte{}, err
	}
	return b, nil
}

func setValue(pd *pb.EdgeChannelData, d ds.EdgeChannelData) {
	switch d.Value.Type {
	case gjson.Null:
		{
			return
		}
	case gjson.True:
		{
			pd.Value = &pb.EdgeChannelData_ValueBool{
				ValueBool: true,
			}
			return
		}
	case gjson.False:
		{
			pd.Value = &pb.EdgeChannelData_ValueBool{
				ValueBool: true,
			}
			return
		}
	case gjson.Number:
		{
			n := d.Value.Num
			if n == math.Trunc(n) {
				pd.Value = &pb.EdgeChannelData_ValueInt64{
					ValueInt64: int64(n),
				}
			} else {
				pd.Value = &pb.EdgeChannelData_ValueDouble{
					ValueDouble: n,
				}
			}
			return
		}
	case gjson.String:
		{
			pd.Value = &pb.EdgeChannelData_ValueString{
				ValueString: d.Value.String(),
			}
			return
		}
	default:
		{
			pd.Value = &pb.EdgeChannelData_ValueBytes{
				ValueBytes: []byte(d.Value.Raw),
			}
			return
		}
	}
}

func compress(buf []byte) ([]byte, error) {
	var cbuf bytes.Buffer
	w, err := xz.NewWriter(&cbuf)
	if err != nil {
		return []byte{}, err
	}
	if _, err := w.Write(buf); err != nil {
		return []byte{}, err
	}
	if err := w.Close(); err != nil {
		return []byte{}, err
	}
	return cbuf.Bytes(), nil
}
